package com.heiye.count.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.github.phantomthief.collection.BufferTrigger;
import com.google.common.collect.Lists;
import com.heiye.count.biz.constants.MQConstants;
import com.heiye.count.biz.constants.RedisKeyConstants;
import com.heiye.count.biz.domain.mapper.QuizCountDOMapper;
import com.heiye.count.biz.model.dto.CountPublishCommentMqDTO;
import com.heiye.framework.common.util.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author: heiye
 * @date: 2025/07/30 下午3:44
 * @version: v1.0.0
 * @description: 计数: 笔记评论数
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "interact_group_" + MQConstants.TOPIC_COUNT_QUIZ_COMMENT,
        topic = MQConstants.TOPIC_COUNT_QUIZ_COMMENT
)
public class CountQuizCommentConsumer implements RocketMQListener<String> {

    @Resource
    private QuizCountDOMapper quizCountDOMapper;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            // 缓存队列的最大容量
            .bufferSize(50000)
            // 一批次最多聚合 1000 条
            .batchSize(1000)
            // 多久聚合一次（1s 一次）
            .linger(Duration.ofSeconds(1))
            // 设置消费者方法
            .setConsumerEx(this::consumeMessage)
            .build();

    @Override
    public void onMessage(String body) {
        // 往 bufferTrigger 中添加元素
        bufferTrigger.enqueue(body);
    }

    private void consumeMessage(List<String> bodys) {
        log.info("==> 【问题评论数】聚合消息, size: {}", bodys.size());
        log.info("==> 【问题评论数】聚合消息, {}", JsonUtils.toJsonString(bodys));

        // 将聚合后的消息体 Json 转 List<CountPublishCommentMqDTO>
        List<CountPublishCommentMqDTO> countPublishCommentMqDTOS = Lists.newArrayList();
        bodys.forEach(body -> {
            try {
                List<CountPublishCommentMqDTO> list = JsonUtils.parseList(body, CountPublishCommentMqDTO.class);
                countPublishCommentMqDTOS.addAll(list);
            } catch (Exception e) {
                log.error("", e);
            }
        });

        // 按问题 ID 进行分组
        Map<Long, List<CountPublishCommentMqDTO>> groupMap = countPublishCommentMqDTOS.stream()
                .collect(Collectors.groupingBy(CountPublishCommentMqDTO::getQuizId));

        // 循环分组字典
        for (Map.Entry<Long, List<CountPublishCommentMqDTO>> entry : groupMap.entrySet()) {
            // 问题 ID
            Long quizId = entry.getKey();
            //  评论数
            int count = CollUtil.size(entry.getValue());

            // 更新 Redis 缓存中的笔记评论总数
            // 构建 Key
            String noteCountHashKey = RedisKeyConstants.buildCountQuizKey(quizId);
            // 判断 Hash 是否存在
            boolean hasKey = redisTemplate.hasKey(noteCountHashKey);

            // 若 Hash 存在
            if (hasKey) {
                // 累加更新
                redisTemplate.opsForHash()
                        .increment(noteCountHashKey, RedisKeyConstants.FIELD_COMMENT_TOTAL, count);
            }

            // 若评论数大于零，则进行更新操作：累加评论总数
            if (count > 0) {
                quizCountDOMapper.insertOrUpdateCommentTotalByQuizId(count, quizId);
            }
        }
    }
}
